package com.zidongxiangxi.rabbit.producer.transaction.listener;

import com.alibaba.fastjson.JSON;
import com.zidongxiangxi.rabbit.core.util.RabbitUtils;
import com.zidongxiangxi.rabbit.producer.entity.Producer;
import com.zidongxiangxi.rabbit.producer.entity.RabbitMessageInfo;
import com.zidongxiangxi.rabbit.producer.manager.ProducerManager;
import com.zidongxiangxi.rabbit.producer.manager.ProducerSequenceManager;
import com.zidongxiangxi.rabbit.producer.support.RabbitSequenceMessagePostProcessor;
import com.zidongxiangxi.rabbit.producer.transaction.support.RabbitMessageTransactionContext;
import com.zidongxiangxi.rabbit.producer.transaction.support.RabbitMessageTransactionHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.lang.Nullable;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Objects;
import java.util.Stack;

/**
 * RabbitMq事务监听
 *
 * @author chenxudong
 * @date 2020/11/07
 */
public class RabbitTransactionListener implements TransactionListener {
	private final Logger logger = LoggerFactory.getLogger(RabbitTransactionListener.class);

	private final ThreadLocal<Stack<RabbitMessageTransactionHolder>> resource =
			ThreadLocal.withInitial(Stack::new);
	private final ProducerManager producerManager;
	private final ProducerSequenceManager producerSequenceManager;
	private final RabbitSequenceMessagePostProcessor rabbitSequenceMessagePostProcessor;

	public RabbitTransactionListener(
			ProducerManager producerManager,
			@Nullable ProducerSequenceManager producerSequenceManager,
			@Nullable RabbitSequenceMessagePostProcessor rabbitSequenceMessagePostProcessor
	) {
		this.producerManager = producerManager;
		this.producerSequenceManager = producerSequenceManager;
		this.rabbitSequenceMessagePostProcessor = rabbitSequenceMessagePostProcessor;
	}

	@Override
	public void beforeCommit() {
		RabbitMessageTransactionHolder messageHolder = RabbitMessageTransactionContext.getMessageHolder();
		if (Objects.isNull(messageHolder) || CollectionUtils.isEmpty(messageHolder.getQueue())) {
			return;
		}
		List<RabbitMessageInfo> list = messageHolder.getQueue();
		for (RabbitMessageInfo messageInfo : list) {
			Producer producer = messageInfo.getProducer();
			try {
				Long recordId = producerManager.saveProducer(messageInfo.getProducer());
				if (Objects.nonNull(producerSequenceManager)
						&& Objects.nonNull(recordId)
						&& RabbitUtils.isSequenceMessage(messageInfo.getMessage())) {
					String messageGroupName = RabbitUtils.getMessageGroupName(messageInfo.getMessage());
					producerSequenceManager.saveProducer(recordId, producer.getApplication(), messageGroupName,
							producer.getMessageId());
				}
			} catch (Throwable e) {
				logger.error("fail to save rabbit producer: " + JSON.toJSONString(messageInfo.getProducer()), e);
			}
		}
	}

	@Override
	public void beforeCompletion() {
		//TODO NOTHING
	}

	@Override
	public void afterCommit() {
		RabbitMessageTransactionHolder messageHolder = RabbitMessageTransactionContext.getMessageHolder();
		RabbitMessageTransactionContext.removeMessageHolder();
		if (Objects.isNull(messageHolder) || CollectionUtils.isEmpty(messageHolder.getQueue())) {
			return;
		}
		for (RabbitMessageInfo messageInfo : messageHolder.getQueue()) {
			try {
				String exchange = messageInfo.getExchange();
				String routingKey = messageInfo.getRoutingKey();
				Message message = messageInfo.getMessage();
				if (Objects.nonNull(rabbitSequenceMessagePostProcessor)) {
					message = rabbitSequenceMessagePostProcessor.postProcessMessage(message);
				}
				CorrelationData correlationData = messageInfo.getCorrelationData();
				messageInfo.getRabbitTemplate().send(exchange, routingKey, message, correlationData);
			} catch (Throwable t) {
				logger.error("fail to send message, [{}]", JSON.toJSONString(messageInfo.getProducer()), t);
			}
		}
	}

	@Override
	public void afterCompletion() {
		RabbitMessageTransactionHolder messageHolder = RabbitMessageTransactionContext.getMessageHolder();
		RabbitMessageTransactionContext.removeMessageHolder();
		if (Objects.isNull(messageHolder) || CollectionUtils.isEmpty(messageHolder.getQueue())) {
			return;
		}
		for (RabbitMessageInfo messageInfo : messageHolder.getQueue()) {
			Producer producer = messageInfo.getProducer();
			logger.info("fail to commit transaction, message({}) is ignored. exchange:{}. routingKey:{}",
					producer.getMessageId(),
					producer.getExchange(), producer.getRoutingKey());
		}
	}

	@Override
	public void suspend() {
		RabbitMessageTransactionHolder messageHolder = RabbitMessageTransactionContext.getMessageHolder();
		if (messageHolder == null) {
			return;
		}
		RabbitMessageTransactionContext.removeMessageHolder();
		resource.get().push(messageHolder);
	}

	@Override
	public void resume() {
		RabbitMessageTransactionContext.setMessageHolder(resource.get().pop());
	}

}
